package com.wakaca.quickstart;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) {

        OrderEventFactory orderEventFactory = new OrderEventFactory();
        int ringBufferSize = 1024*1024;
        // 内部的 LinkedBlockingQueue 没有设置上界，这是一个安全隐患
        // 我们可以自己 new ThreadPoolExecutor 来设定上界，从而规避掉隐患
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        /**
         * 实例化 disruptor 对象
         * 1.orderEventFactory 消息(event)工厂对象
         * 2.ringBufferSize 容器的长度
         * 3.executor 线程池（强制用自定义线程池） RejectedExecutionHandler
         * 4.ProducerType   单生产者 或 多生产者
         * 4.BlockingWaitStrategy   等待策略
         */
        Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(
                orderEventFactory,
                ringBufferSize,
                executor,
                ProducerType.SINGLE, // 单生产者
                new BlockingWaitStrategy()
        );

        /**
         * 添加消费者监听
         * 构建 disruptor 与消费者的一个关联关系
         */
        disruptor.handleEventsWith(new OrderEventHandler());

        /** 启动 disruptor */
        disruptor.start();

        /**
         * 获取实际存储数据的容器：RingBuffer
         */
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        OrderEventProducer producer = new OrderEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long i = 0; i < 100; i++) {
            bb.putLong(0, i);
            producer.sendData(bb);
        }

        disruptor.shutdown();
        executor.shutdown();

    }
}
